package repo

import (
	"context"
	"git.multiverse.io/eventkit/dxc-server/config"
	"git.multiverse.io/eventkit/dxc-server/constant"
	"git.multiverse.io/eventkit/dxc-server/models"
	"git.multiverse.io/framework/log"
	"github.com/go-errors/errors"
	"github.com/go-redis/redis/v8"
	jsoniter "github.com/json-iterator/go"
	"strconv"
	"strings"
	"time"
)

const (
	Standalone = "standalone"
	Cluster    = "cluster"
)

// redis data structure:
// use hash,redis key: `nodeId`:`rootXid`
// hash key: regTm                              	   hash value: RootTxn Register Timestamp
// hash key: txSts                                     hash value: RootTxn Status (0-trying, 1-confirming, 2-confirmed, 3-cancelling, 4-cancelled)
// hash key: stRcTm                      	   		   hash value: RootTxn Start Recovery Timestamp (only recovery use)
// hash key: @`branchXid`                              hash value: branch info (json, TsInfo struct)
// hash key: `branchXid`_sts                           hash value: branch status (0-trying, 1-confirmed, 2-cancelled)
// hash key: `branchXid`_rcTms                  	   hash value: branch recoverTimes
// hash key: `branchXid`_lstUpTm                	   hash value: branch LastUpdateTime (only recover use)
// hash key: `branchXid`_igCancel 				  	   hash value: branch TryFailedIgnoreCallbackCancel
// hash key: hasExpTm                         	  	   hash value: RootTxn HasExpireTime

type RedisRepository struct {
	operate   RedisOperate
	redisType string
	nodeId    string
}

var json = jsoniter.ConfigCompatibleWithStandardLibrary

func NewRedisRepository() *RedisRepository {
	log.Infos("Start initialize redis repository...")
	defer log.Infos("Successfully initialized redis repository!")

	dxcConfig := config.Cfg
	var operate RedisOperate
	if dxcConfig.Repo.Redis.Type == Standalone {
		log.Infos("Start new redis client in standalone mode")
		rdb := redis.NewClient(&redis.Options{
			Addr:     dxcConfig.Repo.Redis.Addr[0],
			Password: dxcConfig.Repo.Redis.Password,
		})
		operate = NewStandaloneRedis(rdb)
	} else {
		log.Infos("Start new redis client in cluster mode")
		rdb := redis.NewClusterClient(&redis.ClusterOptions{
			Addrs:    dxcConfig.Repo.Redis.Addr,
			Password: dxcConfig.Repo.Redis.Password,
		})
		operate = NewClusterRedis(rdb)
	}

	return &RedisRepository{
		operate:   operate,
		redisType: dxcConfig.Repo.Type,
		nodeId:    dxcConfig.NodeId,
	}
}

func (r *RedisRepository) AddRoot(transaction *models.Transaction) (int, error) {
	tsInfo := &TsInfo{
		Xid:       transaction.BranchXid,
		PXid:      transaction.ParentXid,
		PrtAddr:   transaction.ParticipantAddress,
		SrvNm:     transaction.ServiceName,
		Param:     transaction.ParamData,
		StTm:      transaction.BranchTxnStartTime,
		CliSDKVer: transaction.ClientSDKVersion,
		Env:       transaction.Environment,
		Wks:       transaction.Workspace,
		Hder:      transaction.Headers,
	}
	str, err := tsInfo.encode()
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} marshal json fail, err:%s", transaction.RootXid, err)
		return 0, err
	}
	key := r.nodeId + ":" + transaction.BranchXid
	hRegisterTimeKey := constant.CACHE_REGISTER_TIME_KEY
	hRegisterTimeVal := r.timeToString(transaction.BranchTxnStartTime)
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	hTxStatusVal := constant.ROOT_TXN_STATUS_TRYING
	hKey := constant.CACHE_BRANCH_KEY_PREFIX + transaction.BranchXid
	hVal := str
	hStatusKey := transaction.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
	hStatusVal := constant.BRANCH_TXN_STATUS_TRYING
	success, err := r.operate.HMSet(key, hRegisterTimeKey, hRegisterTimeVal, hTxStatusKey, hTxStatusVal, hKey, hVal, hStatusKey, hStatusVal)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} save redis fail, err:%s", transaction.BranchXid, err)
		return 0, err
	}
	if !success {
		err = errors.Errorf("transaction{RootXid:[%s]} save redis fail, return insert false", transaction.BranchXid)
		return 0, err
	}
	return 1, nil
}

func (r *RedisRepository) AddBranch(transaction *models.Transaction) (int, error) {
	tsInfo := &TsInfo{
		Xid:       transaction.BranchXid,
		PXid:      transaction.ParentXid,
		PrtAddr:   transaction.ParticipantAddress,
		SrvNm:     transaction.ServiceName,
		Param:     transaction.ParamData,
		StTm:      transaction.BranchTxnStartTime,
		CliSDKVer: transaction.ClientSDKVersion,
		Env:       transaction.Environment,
		Wks:       transaction.Workspace,
		Hder:      transaction.Headers,
	}
	str, err := tsInfo.encode()
	if err != nil {
		err = errors.Errorf("transaction{BranchXid:[%s]} marshal json fail, err:%s", transaction.RootXid, err)
		return 0, err
	}
	hKey := constant.CACHE_BRANCH_KEY_PREFIX + transaction.BranchXid
	hVal := str
	key := r.nodeId + ":" + transaction.RootXid
	hStatusKey := transaction.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
	hStatusVal := constant.BRANCH_TXN_STATUS_TRYING
	success, err := r.operate.HMSet(key, hKey, hVal, hStatusKey, hStatusVal)
	if err != nil {
		err = errors.Errorf("transaction{BranchXid:[%s]} save redis fail, err:%s", transaction.RootXid, err)
		return 0, err
	}
	if !success {
		err = errors.Errorf("transaction{BranchXid:[%s]} save redis fail, return insert false", transaction.RootXid)
		return 0, err
	}
	return 1, nil
}

func (r *RedisRepository) Recovery(root *models.RootTransaction) (int, error) {
	txns := make([]interface{}, 0)
	hRegisterTimeKey := constant.CACHE_REGISTER_TIME_KEY
	hRegisterTimeVal := r.timeToString(root.RootTxnRegisterTime)
	txns = append(txns, hRegisterTimeKey)
	txns = append(txns, hRegisterTimeVal)
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	hTxStatusVal := root.RootTxnStat
	txns = append(txns, hTxStatusKey)
	txns = append(txns, hTxStatusVal)
	for _, branch := range root.Branches {
		tsInfo := &TsInfo{
			Xid:       branch.BranchXid,
			PXid:      branch.ParentXid,
			PrtAddr:   branch.ParticipantAddress,
			SrvNm:     branch.ServiceName,
			Param:     branch.ParamData,
			StTm:      branch.BranchTxnStartTime,
			CliSDKVer: branch.ClientSDKVersion,
			Env:       branch.Environment,
			Wks:       branch.Workspace,
		}
		str, err := json.MarshalToString(tsInfo)
		if err != nil {
			err = errors.Errorf("transaction{RootXid:[%s]} marshal json fail, err:%s", branch.RootXid, err)
			return 0, err
		}
		hKey := constant.CACHE_BRANCH_KEY_PREFIX + branch.BranchXid
		hVal := str
		hStatusKey := branch.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
		hStatusVal := branch.BranchTxnStat
		txns = append(txns, hKey)
		txns = append(txns, hVal)
		txns = append(txns, hStatusKey)
		txns = append(txns, hStatusVal)
	}
	key := r.nodeId + ":" + root.RootXid
	success, err := r.operate.HMSet(key, txns...)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} save redis fail, err:%s", root.RootXid, err)
		return 0, err
	}
	if !success {
		err = errors.Errorf("transaction{RootXid:[%s]} save redis fail, return insert false", root.RootXid)
		return 0, err
	}
	return 1, nil
}

func (r *RedisRepository) FindAllBranchByRootXid(rootXid string) (*models.RootTransaction, error) {
	key := r.nodeId + ":" + rootXid
	mp, err := r.operate.HGetAll(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} redis HGetAll fail, err:%s", rootXid, err)
		return nil, err
	}
	if len(mp) == 0 {
		err = errors.Errorf("transaction{RootXid:[%s]} this transaction not exist", rootXid)
		return nil, err
	}
	hRegisterTimeKey := constant.CACHE_REGISTER_TIME_KEY
	registerTime, err := r.stringToTime(mp[hRegisterTimeKey])
	if err != nil {
		return nil, err
	}
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	var startRecoverTime time.Time
	hStartRecoverTimeKey := constant.CACHE_START_RECOVER_TIME_KEY
	if startRecoverTimeStr, ok := mp[hStartRecoverTimeKey]; ok {
		startRecoverTime, err = r.stringToTime(startRecoverTimeStr)
		if err != nil {
			return nil, err
		}
	}
	var hasExpireTime bool
	hHasExpireTimeKey := constant.CACHE_HAS_EXPIRE_TIME_KEY
	if hHasExpireTimeVal, ok := mp[hHasExpireTimeKey]; ok {
		hasExpireTime, err = strconv.ParseBool(hHasExpireTimeVal)
		if err != nil {
			return nil, err
		}
	}
	rootTransaction := &models.RootTransaction{
		RootXid:                 rootXid,
		RootTxnStat:             mp[hTxStatusKey],
		RootTxnRegisterTime:     registerTime,
		RootTxnStartRecoverTime: startRecoverTime,
		HasExpireTime:           hasExpireTime,
	}
	// struct branches
	branchXIds := make([]string, 0)
	for k := range mp {
		if strings.HasPrefix(k, constant.CACHE_BRANCH_KEY_PREFIX) {
			branchXIds = append(branchXIds, k[constant.LENGTH_OF_CACHE_BRANCH_KEY_PREFIX:])
		}
	}
	transactions := make([]*models.Transaction, 0)
	for _, branchXId := range branchXIds {
		hBranchStatusKey := branchXId + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
		hRecoverTimesKey := branchXId + constant.CACHE_BRANCH_RECOVER_TIMES_SUFFIX_KEY
		recoveryTimes := 0
		if _, ok := mp[hBranchStatusKey]; ok {
			recoveryTimes, _ = strconv.Atoi(mp[hRecoverTimesKey])
		}
		tsInfo := new(TsInfo)
		tsInfoStr, ok := mp[constant.CACHE_BRANCH_KEY_PREFIX+branchXId]
		if !ok {
			return nil, errors.Errorf("transaction{RootXid:[%s], BranchXid:[%s]} this branch transaction not exist", rootXid, branchXId)
		}

		if err := tsInfo.decode([]byte(tsInfoStr)); err != nil {
			err = errors.Errorf("transaction{RootXid:[%s], BranchXid:[%s]} Unmarshal json fail, err:%s, mp[branchXId]=[%s]", rootXid, branchXId, err, mp[branchXId])
			return nil, err
		}
		transaction := &models.Transaction{
			RootXid:            rootXid,
			ParentXid:          tsInfo.PXid,
			BranchXid:          branchXId,
			ParticipantAddress: tsInfo.PrtAddr,
			BranchTxnStat:      mp[hBranchStatusKey],
			RecoveryTimes:      recoveryTimes,
			ServiceName:        tsInfo.SrvNm,
			ParamData:          tsInfo.Param,
			BranchTxnStartTime: tsInfo.StTm,
			ClientSDKVersion:   tsInfo.CliSDKVer,
			Environment:        tsInfo.Env,
			Workspace:          tsInfo.Wks,
			Headers:            tsInfo.Hder,
		}
		hLastUpdateTimeKey := branchXId + constant.CACHE_BRANCH_LAST_UPDATE_TIME_SUFFIX_KEY

		if lastUpdateTimeStr, ok := mp[hLastUpdateTimeKey]; ok {
			lastUpdateTime, err := r.stringToTime(lastUpdateTimeStr)
			if err != nil {
				return nil, err
			}
			transaction.BranchTxnLastUpdateTime = lastUpdateTime
		}
		hTryFailedIgnoreCallbackCancelKey := branchXId + constant.CACHE_BRANCH_TRY_FAILED_IGNORE_CALLBACK_CANCEL_SUFFIX_KEY
		if tryFailedIgnoreCallbackCancelStr, ok := mp[hTryFailedIgnoreCallbackCancelKey]; ok {
			tryFailedIgnoreCallbackCancel, err := strconv.ParseBool(tryFailedIgnoreCallbackCancelStr)
			if err != nil {
				return nil, err
			}
			transaction.TryFailedIgnoreCallbackCancel = tryFailedIgnoreCallbackCancel
		}
		transactions = append(transactions, transaction)
	}
	rootTransaction.Branches = transactions
	return rootTransaction, nil
}

func (r *RedisRepository) DeleteTxn(rootXid string) (int, error) {
	key := r.nodeId + ":" + rootXid
	count, err := r.operate.Del(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} delete txn redis Del fail, err:%s", rootXid, err)
		return 0, err
	}
	strInt64 := strconv.FormatInt(count, 10)
	result, _ := strconv.Atoi(strInt64)
	return result, nil
}

func (r *RedisRepository) IsExistsRootTransaction(rootXid string) (bool, error) {
	key := r.nodeId + ":" + rootXid
	exists, err := r.operate.Exists(key)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} exist root tx redis HDel fail, err:%s", rootXid, err)
		return false, err
	}
	return exists, nil
}

func (r *RedisRepository) IsExistsBranchTransaction(rootXid string, branchXid string) (bool, error) {
	key := r.nodeId + ":" + rootXid
	exists, err := r.operate.HExist(key, constant.CACHE_BRANCH_KEY_PREFIX+branchXid)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s],BranchXid: [%s]} HExist fail, err:%s", rootXid, branchXid, err)
		return false, err
	}
	return exists, nil
}

// modify root tx status and return root Transaction info
func (r *RedisRepository) TryCallback(rootXid string, callbackType string) (result *models.RootTransaction, err error) {
	// check exist
	exists, err := r.IsExistsRootTransaction(rootXid)
	if err != nil {
		return nil, err
	}
	if !exists {
		err = errors.Errorf("Register transaction{RootXid:[%s]} this root transaction not exist", rootXid)
		return nil, err
	}
	key := r.nodeId + ":" + rootXid
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	success, err := r.operate.HSet(key, hTxStatusKey, callbackType)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s], callbackType:[%s]} set root tx status redis HSet fail, err:%s", rootXid, callbackType, err)
		return nil, err
	}
	if !success {
		err = errors.Errorf("transaction{RootXid:[%s], callbackType:[%s]} set root tx status redis HSet fail, return false", rootXid, callbackType)
		return nil, err
	}
	result, err = r.FindAllBranchByRootXid(rootXid)
	if err != nil {
		return nil, err
	}
	return result, nil
}

// check root tx exist & status
func (r *RedisRepository) TryEnList(rootXid string) error {
	key := r.nodeId + ":" + rootXid
	hTxStatusKey := constant.CACHE_TX_STATUS_KEY
	status, err := r.operate.HGet(key, hTxStatusKey)
	if err != nil {
		err = errors.Errorf("transaction{RootXid:[%s]} get root tx status redis HGet fail, err:%s", rootXid, err)
		return err
	}
	if status != "0" {
		return errors.Errorf("TryEnList failed, Invalid root transaction stat[%s]", rootXid)
	}
	return nil
}

// update branch RecoveryTimes、BranchTxnStat、BranchTxnLastUpdateTime && transaction start_recover_time
func (r *RedisRepository) CallbackFail(rootXid string, branches []*models.Transaction, firstTimeCallback bool) (int, error) {
	key := r.nodeId + ":" + rootXid
	values := make([]string, 0)
	if firstTimeCallback {
		hStartRecoverTimeKey := constant.CACHE_START_RECOVER_TIME_KEY
		hStartRecoverTimeVal := r.timeToString(time.Now())
		values = append(values, hStartRecoverTimeKey)
		values = append(values, hStartRecoverTimeVal)
	}
	for _, branch := range branches {
		hStatusKey := branch.BranchXid + constant.CACHE_BRANCH_STATUS_SUFFIX_KEY
		hStatusVal := branch.BranchTxnStat
		hRecoverTimesKey := branch.BranchXid + constant.CACHE_BRANCH_RECOVER_TIMES_SUFFIX_KEY
		hRecoverTimesVal := strconv.Itoa(branch.RecoveryTimes)
		hLastUpdateTimeKey := branch.BranchXid + constant.CACHE_BRANCH_LAST_UPDATE_TIME_SUFFIX_KEY
		hLastUpdateTimeVal := strconv.FormatInt(branch.BranchTxnLastUpdateTime.Unix(), 10)
		values = append(values, hStatusKey)
		values = append(values, hStatusVal)
		values = append(values, hRecoverTimesKey)
		values = append(values, hRecoverTimesVal)
		values = append(values, hLastUpdateTimeKey)
		values = append(values, hLastUpdateTimeVal)
	}
	success, err := r.operate.HMSet(key, values)
	if err != nil {
		return 0, err
	}
	if !success {
		return 0, errors.Errorf("transaction{RootXid:[%s]} CallbackFail redis fail", rootXid)
	}
	return len(branches), nil
}

func (r *RedisRepository) FindRootBranchesByTimeExpire(expireSeconds int64) (rootXIds []*models.Transaction, err error) {
	// todo
	return nil, nil
}

func (r *RedisRepository) UpdateBranchTryFailedIgnoreCallbackCancel(rootXid string, branchXid string) (int, error) {
	key := r.nodeId + ":" + rootXid
	hTryFailedIgnoreCallbackCancelKey := branchXid + constant.CACHE_BRANCH_TRY_FAILED_IGNORE_CALLBACK_CANCEL_SUFFIX_KEY
	hTryFailedIgnoreCallbackCancelVal := strconv.FormatBool(true)
	values := make([]string, 0)
	values = append(values, hTryFailedIgnoreCallbackCancelKey)
	values = append(values, hTryFailedIgnoreCallbackCancelVal)
	success, err := r.operate.HMSet(key, values)
	if err != nil {
		return 0, err
	}
	if !success {
		return 0, errors.Errorf("transaction{RootXid:[%s], BranchXid:[%s]} UpdateBranchTryFailedIgnoreCallbackCancel redis fail", rootXid, branchXid)
	}
	return 1, nil
}

func (r *RedisRepository) ExpireTime(rootXid string) (int, error) {
	key := r.nodeId + ":" + rootXid
	hHasExpireTimeKey := constant.CACHE_HAS_EXPIRE_TIME_KEY
	hHasExpireTimeVal := strconv.FormatBool(true)
	values := make([]string, 0)
	values = append(values, hHasExpireTimeKey)
	values = append(values, hHasExpireTimeVal)
	success, err := r.operate.HMSet(key, values)
	if err != nil {
		return 0, err
	}
	if !success {
		return 0, errors.Errorf("transaction{RootXid:[%s]} ExpireTime redis fail", rootXid)
	}
	return 1, nil
}

func (r *RedisRepository) LoadRootsFromPersistenceRepo(count int64) ([]string, error) {
	pattern := r.nodeId + ":" + "*"
	return r.operate.ScanAll(0, pattern, count)
}

func (r *RedisRepository) timeToString(t time.Time) string {
	return strconv.FormatInt(t.Unix(), 10)
}

func (r *RedisRepository) stringToTime(s string) (time.Time, error) {
	timestamp, err := strconv.ParseInt(s, 10, 64)
	if err != nil {
		err = errors.Errorf("string to time fail, err:%s", err)
	}
	return time.Unix(timestamp, 0), nil
}

type RedisOperate interface {
	EXPIRE(key string, duration time.Duration) (bool, error)
	Exists(keys ...string) (bool, error)
	HExist(key string, hKey string) (bool, error)
	HMSet(key string, values ...interface{}) (bool, error)
	HSet(key string, values ...interface{}) (bool, error)
	HGetAll(key string) (map[string]string, error)
	HGet(key string, hKey string) (string, error)
	HMGet(key string, hKeys ...string) ([]string, error)
	Del(keys ...string) (int64, error)
	HDel(key string, hKeys ...string) (int64, error)
	Ping() (string, error)
	ScanAll(cursor uint64, match string, count int64) ([]string, error)
}

type StandaloneRedis struct {
	client *redis.Client
}

func NewStandaloneRedis(client *redis.Client) *StandaloneRedis {
	return &StandaloneRedis{client: client}
}

func (s *StandaloneRedis) EXPIRE(key string, duration time.Duration) (bool, error) {
	return s.client.Expire(context.Background(), key, duration).Result()
}

func (s *StandaloneRedis) Exists(keys ...string) (bool, error) {
	result, err := s.client.Exists(context.Background(), keys...).Result()
	if err != nil {
		return false, err
	}
	return result == 1, nil
}

func (s *StandaloneRedis) HExist(key string, hKey string) (bool, error) {
	return s.client.HExists(context.Background(), key, hKey).Result()
}

func (s *StandaloneRedis) HMSet(key string, values ...interface{}) (bool, error) {
	return s.client.HMSet(context.Background(), key, values...).Result()
}

func (s *StandaloneRedis) HSet(key string, values ...interface{}) (bool, error) {
	result, err := s.client.HSet(context.Background(), key, values...).Result()
	if err != nil {
		return false, err
	}
	if result == 0 || result == 1 {
		return true, nil
	}
	return false, nil
}

func (s *StandaloneRedis) HGetAll(key string) (map[string]string, error) {
	return s.client.HGetAll(context.Background(), key).Result()
}

func (s *StandaloneRedis) Del(keys ...string) (int64, error) {
	return s.client.Del(context.Background(), keys...).Result()
}

func (s *StandaloneRedis) HGet(key string, hKey string) (string, error) {
	return s.client.HGet(context.Background(), key, hKey).Result()
}

func (s *StandaloneRedis) HMGet(key string, hKeys ...string) ([]string, error) {
	results, err := s.client.HMGet(context.Background(), key, hKeys...).Result()
	if err != nil {
		return nil, err
	}
	slice := make([]string, len(results))
	for idx, result := range results {
		r := result.(string)
		slice[idx] = r
	}
	return slice, nil
}

func (s *StandaloneRedis) HDel(key string, hKeys ...string) (int64, error) {
	return s.client.HDel(context.Background(), key, hKeys...).Result()
}

func (s *StandaloneRedis) Ping() (string, error) {
	return s.client.Ping(context.Background()).Result()
}

func (s *StandaloneRedis) ScanAll(cursor uint64, match string, count int64) ([]string, error) {
	result := make([]string, 0)

	for {
		keys, idx, err := s.client.Scan(context.Background(), cursor, match, count).Result()
		if err != nil {
			return nil, err
		}
		result = append(result, keys...)
		if idx == 0 {
			break
		}
		cursor = idx
	}
	return result, nil
}

type ClusterRedis struct {
	client *redis.ClusterClient
}

func NewClusterRedis(client *redis.ClusterClient) *ClusterRedis {
	return &ClusterRedis{client: client}
}

func (c *ClusterRedis) EXPIRE(key string, duration time.Duration) (bool, error) {
	return c.client.Expire(context.Background(), key, duration).Result()
}

func (c *ClusterRedis) Exists(keys ...string) (bool, error) {
	result, err := c.client.Exists(context.Background(), keys...).Result()
	if err != nil {
		return false, err
	}
	return result == 1, nil
}

func (c *ClusterRedis) HExist(key string, hKey string) (bool, error) {
	return c.client.HExists(context.Background(), key, hKey).Result()
}

func (c *ClusterRedis) HMSet(key string, values ...interface{}) (bool, error) {
	return c.client.HMSet(context.Background(), key, values...).Result()
}

func (c *ClusterRedis) HSet(key string, values ...interface{}) (bool, error) {
	result, err := c.client.HSet(context.Background(), key, values...).Result()
	if err != nil {
		return false, err
	}
	if result == 0 || result == 1 {
		return true, nil
	}
	return false, nil
}

func (c *ClusterRedis) HGetAll(key string) (map[string]string, error) {
	return c.client.HGetAll(context.Background(), key).Result()
}

func (c *ClusterRedis) Del(keys ...string) (int64, error) {
	return c.client.Del(context.Background(), keys...).Result()
}

func (c *ClusterRedis) HGet(key string, hKey string) (string, error) {
	return c.client.HGet(context.Background(), key, hKey).Result()
}

func (c *ClusterRedis) HMGet(key string, hKeys ...string) ([]string, error) {
	results, err := c.client.HMGet(context.Background(), key, hKeys...).Result()
	if err != nil {
		return nil, err
	}
	slice := make([]string, len(results))
	for idx, result := range results {
		r := result.(string)
		slice[idx] = r
	}
	return slice, nil
}

func (c *ClusterRedis) HDel(key string, hKeys ...string) (int64, error) {
	return c.client.HDel(context.Background(), key, hKeys...).Result()
}

func (c *ClusterRedis) Ping() (string, error) {
	return c.client.Ping(context.Background()).Result()
}

func (c *ClusterRedis) ScanAll(cursor uint64, match string, count int64) ([]string, error) {
	result := make([]string, 0)
	err := c.client.ForEachMaster(context.Background(), func(ctx context.Context, client *redis.Client) error {
		for {
			keys, idx, err := client.Scan(context.Background(), cursor, match, count).Result()
			if err != nil {
				return err
			}
			result = append(result, keys...)
			if idx == 0 {
				break
			}
			cursor = idx
		}
		return nil
	})
	return result, err
}
